package com.cctc.rds.nrpc.core.server;

import com.cctc.rds.nrpc.api.connection.ConnectionPool;
import com.cctc.rds.nrpc.api.rpc.RpcServerRouter;
import com.cctc.rds.nrpc.api.server.RpcServer;
import com.cctc.rds.nrpc.api.server.config.RpcServerConfig;
import com.cctc.rds.nrpc.core.protocol.codec.MessageDecode;
import com.cctc.rds.nrpc.core.protocol.codec.MessageEncode;
import com.cctc.rds.nrpc.core.protocol.codec.MessageUnpacket;
import com.cctc.rds.nrpc.core.server.connnection.DefaultServerConnectionPool;
import com.cctc.rds.nrpc.core.server.connnection.ServerConnectionStatus;
import com.cctc.rds.nrpc.core.server.handler.ServerIdleStateHandler;
import com.cctc.rds.nrpc.core.server.handler.ServerRpcMessageHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.GenericFutureListener;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class NettyRpcServer implements RpcServer {
    private NioEventLoopGroup bossGroup, workerGroup;
    private RpcServerConfig config;
    private DefaultServerConnectionPool connectionPool;
    private RpcServerRouter router;
    private final AtomicInteger status = new AtomicInteger();
    private final AtomicInteger failCount = new AtomicInteger();
    public static final int MAX_RETRY_COUNT = 3, RETRY_SLEEP_MS_UNIT = 2 * 1000;

    public NettyRpcServer(RpcServerRouter router) {
        this.router = router;
        this.status.set(ServerConnectionStatus.OFF.getValue());
        this.failCount.set(0);
    }

    @Override
    public void start(RpcServerConfig config) {
        this.config = config;
        start();
    }

    private void start() {
        if (!status.compareAndSet(ServerConnectionStatus.OFF.getValue(), ServerConnectionStatus.STARTING.getValue()))
            return;
        ServerBootstrap bootstrap = new ServerBootstrap();
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
        bootstrap
                .group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childOption(ChannelOption.SO_SNDBUF, 32 * 1024)
                .childOption(ChannelOption.SO_RCVBUF, 32 * 1024)
                .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) config.getIdleStateMs())
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new MessageUnpacket())
                                .addLast(new MessageDecode())
                                .addLast(new MessageEncode())
                                .addLast(new ServerIdleStateHandler(config.getIdleStateMs()))
                                .addLast(new ServerRpcMessageHandler(router, connectionPool));
                    }
                })
                .bind(config.getPort())
                .addListener((GenericFutureListener<ChannelFuture>) future -> {
                    if (future.cause() != null) {
                        status.compareAndSet(ServerConnectionStatus.STARTING.getValue(), ServerConnectionStatus.OFF.getValue());
                        int count = failCount.incrementAndGet();
                        if (count > MAX_RETRY_COUNT) return;
                        try {
                            TimeUnit.MILLISECONDS.sleep(count * RETRY_SLEEP_MS_UNIT);
                            start();
                        } catch (InterruptedException ignored) {
                        }
                    } else {
                        status.compareAndSet(ServerConnectionStatus.STARTING.getValue(), ServerConnectionStatus.STARTED.getValue());
                    }
                });
    }

    @Override
    public ConnectionPool getConnectionPool() {
        return connectionPool;
    }

    @Override
    public void stop() {
        while (status.get() == ServerConnectionStatus.STARTING.getValue()) {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException ignored) {
            }
        }
        if (status.compareAndSet(ServerConnectionStatus.STARTED.getValue(), ServerConnectionStatus.OFF.getValue())) {
            try {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
                connectionPool.shutdown();
                failCount.set(0);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}
